package com.cskaoyan.mall.promo.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.cskaoyan.mall.common.constant.RedisConst;
import com.cskaoyan.mall.common.execption.BusinessException;
import com.cskaoyan.mall.common.util.DateUtil;
import com.cskaoyan.mall.mq.constant.MqResultEnum;
import com.cskaoyan.mall.mq.constant.MqTopicConst;
import com.cskaoyan.mall.order.dto.OrderDetailDTO;
import com.cskaoyan.mall.order.dto.OrderTradeDTO;
import com.cskaoyan.mall.order.query.OrderInfoParam;
import com.cskaoyan.mall.promo.api.dto.SeckillGoodsDTO;
import com.cskaoyan.mall.promo.client.OrderApiClient;
import com.cskaoyan.mall.promo.client.UserApiClient;
import com.cskaoyan.mall.promo.constant.LocalStockStatus;
import com.cskaoyan.mall.promo.constant.SeckillCodeEnum;
import com.cskaoyan.mall.promo.constant.SeckillGoodsStatus;
import com.cskaoyan.mall.promo.converter.SeckillGoodsConverter;
import com.cskaoyan.mall.promo.mapper.SeckillGoodsMapper;
import com.cskaoyan.mall.promo.model.SeckillGoods;
import com.cskaoyan.mall.promo.mq.PromoTransactionProducer;
import com.cskaoyan.mall.promo.service.PromoService;
import com.cskaoyan.mall.promo.util.LocalCacheHelper;
import com.cskaoyan.mall.user.dto.UserAddressDTO;
import org.redisson.api.RMap;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.StringCodec;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @Author Pudding
 * @Date 2024/7/22 11:30
 */
@Service
public class PromoServiceImpl implements PromoService {

    @Autowired
    UserApiClient userApiClient;
    @Autowired
    SeckillGoodsConverter seckillGoodsConverter;
    @Autowired
    SeckillGoodsMapper seckillGoodsMapper;
    @Autowired
    OrderApiClient orderApiClient;
    @Autowired
    PromoTransactionProducer promoTransactionProducer;
    @Autowired
    RedissonClient redissonClient;

    /**
     * 1. 把商品库存状态位导入到本地JVM的map缓存中
     */
    @Override
    public void importIntoCache() {
        // 1. 从数据库查询今天的秒杀商品数据
        QueryWrapper<SeckillGoods> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("status", "CHECKED_PASS")
                .gt("stock_count", 0)
                .eq("DATE_FORMAT(start_time,'%Y-%m-%d')", DateUtil.formatDate(new Date()));
        List<SeckillGoods> seckillGoodsList = seckillGoodsMapper.selectList(queryWrapper);

        // 2. 初始化每一个商品库存状态位
        for (SeckillGoods seckillGoods : seckillGoodsList) {
            // eg: 50-1 表示skuId为50的商品还有库存
            // eg: 49-0 表示skuId为49的商品没有库存
            LocalCacheHelper.put(seckillGoods.getSkuId().toString(), LocalStockStatus.HAS_STOCK.getNum());
        }

        //3. 缓存预热，提前在定时任务执行的时候，就把商品数据放入缓存中
        // 那么采用什么数据结构来存储一个商品列表呢? ——> 也可以使用hash来存储
        RMap<Long, SeckillGoods> map = redissonClient.getMap(RedisConst.PROMO_SECKILL_GOODS);
        for (SeckillGoods seckillGoods : seckillGoodsList) {
            if (!map.containsKey(seckillGoods.getSkuId())){ //本地测试的时候，为了避免覆盖商品数据库中已经存在的商品数据
                map.put(seckillGoods.getSkuId(), seckillGoods);
            }
        }

        //4. 把商品的库存也存储到redis中
        //使用hash来存储库存
            // key: promo:goods:stock:
                // field: skuId  商品id
                // value: stock  商品的库存

        //说明：在存储库存的时候，不使用JsonJackson的序列化方式
        //因为使用这种方式来存储库存，那么后续使用Lua脚本来扣减库存是找不到对应的商品id的，那么每次扣减库存就会失败，Lua脚本会返回-1
        //所以需要指定库存存储的序列化方式为 String
        RMap<String, String> stockMap = redissonClient.getMap(RedisConst.PROMO_SECKILL_GOODS_STOCK, new StringCodec());
        for (SeckillGoods seckillGoods : seckillGoodsList) {
            Long skuId = seckillGoods.getSkuId();   //商品id
            Integer stockCount = seckillGoods.getStockCount(); //库存数量

            if (!stockMap.containsKey(skuId.toString())){     //本地测试的时候，为了避免覆盖商品数据库中已经存在的商品数据
                stockMap.put(skuId.toString(), stockCount.toString());
            }
        }
    }


    /**
     * 查询当天的秒杀商品列表
     * @return
     */
    private List<SeckillGoods> getCurrentDaySeckillGoods() {
        // where status = CHECKED_PASS AND stock_count > 0 AND DATE_FORMAT(start_time,'%Y-%m-%d') = ?
        /*
        QueryWrapper<SeckillGoods> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("status", "CHECKED_PASS")
                .gt("stock_count", 0)
                .eq("DATE_FORMAT(start_time,'%Y-%m-%d')", DateUtil.formatDate(new Date()));

        List<SeckillGoods> seckillGoodsList = seckillGoodsMapper.selectList(queryWrapper);
        */

        //改进：因为当前已经把商品数据，提前存入了缓存中，所以在此处，可以直接从缓存里面获取数据
        RMap<Long, SeckillGoods> map = redissonClient.getMap(RedisConst.PROMO_SECKILL_GOODS);

        //将map中的value收集为list
        List<SeckillGoods> seckillGoodsList = map.values().stream().collect(Collectors.toList());

        return seckillGoodsList;
    }


    /**
     * 返回全部的秒杀商品列表
     * @return
     */
    @Override
    public List<SeckillGoodsDTO> findAll() {

        // 调用promoService的方法，根据如下条件查询
        // 1. 审核通过：status = CHECKED_PASS
        // 2. 今天开始：DATE_FORMAT(start_time,'%Y-%m-%d') = new Date()
        // 3. 库存大于0：stock_count>0
        List<SeckillGoods> currentDaySeckillGoods = this.getCurrentDaySeckillGoods();
        List<SeckillGoodsDTO> seckillGoodsDTOS = seckillGoodsConverter.convertSeckillGoodsList(currentDaySeckillGoods);

        return seckillGoodsDTOS;
    }


    /**
     * 根据商品id获取秒杀商品详情的实体
     * @param skuId
     * @return
     */
    @Override
    public SeckillGoodsDTO getSeckillGoodsDTO(Long skuId) {
        /*
        for (SeckillGoods currentDaySeckillGood : this.getCurrentDaySeckillGoods()) {
            if (skuId.equals(currentDaySeckillGood.getSkuId())){
                SeckillGoodsDTO seckillGoodsDTO = seckillGoodsConverter.convertSeckillGoodsToDTO(currentDaySeckillGood);
                return seckillGoodsDTO;
            }
        }
        */

        //改进，从Redis中根据skuId获取商品数据
        RMap<Long, SeckillGoods> map = redissonClient.getMap(RedisConst.PROMO_SECKILL_GOODS);
        SeckillGoods seckillGoods = map.get(skuId);

        //进行转换将PO对象转换为DTO对象
        SeckillGoodsDTO seckillGoodsDTO = seckillGoodsConverter.convertSeckillGoodsToDTO(seckillGoods);

        return seckillGoodsDTO;
    }


    /**
     * 根据userId和skuId检查秒杀订单是否已经生成
     * @param skuId
     * @param userId
     * @return
     */
    @Override
    public boolean checkOrder(Long skuId, String userId) {

        //1. 从redis拿取订单生成与否的标记
        RMap<String, Long> map = redissonClient.getMap(RedisConst.PROMO_SECKILL_ORDERS);
        String key = userId + ":" + skuId;

        //2. 对标记进行判断
        //2.1 如果检查到了，说明用户已经下单成功，并且订单已经生成了。返回true。如果没有检查到，返回false
        boolean ret = map.containsKey(key);
        return ret;
    }


    /**
     * 每天秒杀活动结束，清除对应的缓存
     */
    @Override
    public void clearRedisCache() {

        //1. 修改秒杀商品的状态为已结束
        List<SeckillGoods> seckillGoodsList = this.getCurrentDaySeckillGoods();
        for (SeckillGoods seckillGoods : seckillGoodsList) {
            seckillGoods.setStatus(SeckillGoodsStatus.FINISHED.name());
            seckillGoodsMapper.updateById(seckillGoods);
        }

        //2. 更新redis中的库存到数据库中，需要指定序列化方式
        RMap<String, String> map = redissonClient.getMap(RedisConst.PROMO_SECKILL_GOODS_STOCK, new StringCodec());
        map.keySet().stream().forEach(skuId -> {
            String remainStock = map.get(skuId);
            //更新到数据库中
            UpdateWrapper<SeckillGoods> updateWrapper = new UpdateWrapper<>();
            updateWrapper.eq("sku_id", skuId)
                    .eq("DATE_FORMAT(start_time,'%Y-%m-%d')", DateUtil.formatDate(new Date()))
                    .set("stock_count", remainStock);
            seckillGoodsMapper.update(null, updateWrapper);
        });

        //3. 删除对应的缓存（redis中的缓存、本地缓存）
        //3.1 删除redis中的缓存
        redissonClient.getKeys().deleteByPattern("promo:*");

        //3.2 删除本地JVM缓存
        LocalCacheHelper.removeAll();
    }


    /**
     * 获取秒杀商品订单结算/确认页信息
     * @param userId
     * @param seckillGoods
     * @return
     */
    @Override
    public OrderTradeDTO getTradeData(String userId, SeckillGoods seckillGoods) {

        //1. 查询用户收获地址列表
        List<UserAddressDTO> userAddressDTOList = userApiClient.findUserAddressListByUserId(userId);

        //2. 获取订单详情列表
        OrderDetailDTO orderDetailDTO = seckillGoodsConverter.secondKillGoodsToOrderDetailDTO(seckillGoods, 1);
        List<OrderDetailDTO> detailArrayList = Arrays.asList(orderDetailDTO);

        //3. 组装数据
        OrderTradeDTO orderTradeDTO = new OrderTradeDTO();
        orderTradeDTO.setUserAddressList(userAddressDTOList);      //设置用户收获地址列表
        orderTradeDTO.setDetailArrayList(detailArrayList);         //设置商品列表
        orderTradeDTO.setTotalNum(1);                              //设置商品数量
        orderTradeDTO.setTotalAmount(seckillGoods.getCostPrice()); //设置商品秒杀价格

        return orderTradeDTO;
    }

    /**
     * 提交秒杀订单，这个方法没有使用分布式事务控制，所以现在我们不使用了
     * @param orderInfoParam
     * @return
     */
    @Override
    //rollbackFor：什么样的异常就回滚；propagation是spring的传播行为，默认是REQUIRED
    @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
    public boolean submitOrder(OrderInfoParam orderInfoParam) {

        // 1. 扣减秒杀商品库存
        // 1.1 先查询秒杀商品库存
        Long skuId = orderInfoParam.getOrderDetailList().get(0).getSkuId();
        SeckillGoodsDTO seckillGoodsDTO = this.getSeckillGoodsDTO(skuId);
        if (seckillGoodsDTO == null){
            return false;
        }

        // 1.2 给数据库中的库存-1
        int affectedRows = seckillGoodsMapper.decreaseStock(seckillGoodsDTO.getId(), 1);
        if (affectedRows == 0){
            //说明该秒杀商品的库存已经扣减完了，所以需要我们进行更新本地JVM缓存中库存状态位标记为0，表示库存已经没有
            LocalCacheHelper.put(skuId.toString(), LocalStockStatus.NO_STOCK);
            return false;
        }

        /**
         * 由于以上的2步操作，不是一个原子操作，那么在多线程的情况下，就会有数据安全的问题，那么就可能导致【超卖】
         *      比如现在两个线程A，B同时过来访问，现在数据库中该商品还剩余一个库存
         *      A线程和B线程同时判断该商品有没有库存，判断结果是A和B线程都判断有库存
         *      接下来A和B都给库存-1（在内存中） A-1 = 0  B-1=0
         *      A线程把数据库中的库存设置为0，B线程也把数据库中的库存设置为0
         *      相当于A和B线程都扣减了库存，都生成了订单，但是数据库中的库存只扣减了1个
         *      这样就可能导致生成的订单个数超过库存的个数，导致超卖
         * 而为了解决超卖问题，需要【加锁】
         * 方式一：采用数据库的加锁方案。而数据库的增删改操作其实会默认加一个行锁，锁住这一行数据
         *          补充： 如果数据库针对查询操作，也想加行锁，该怎么办呢？
         *                select * from seckill_goods where id = 1 for update;     // 显式的声明，给查询语句增加行锁
         */


        // 2. 调用订单服务，生成秒杀订单，返回orderId
        /*
           问题：远程调用无法通过本地事务注解来解决事务原子性的问题，因为两个服务基于不同的连接
           解决方式：使用分布式事务
        */
        Long orderId = orderApiClient.submitOrder(orderInfoParam);
        if (orderId == null){
            return false;
        }

        return true;
    }


    /**
     * 提交秒杀订单，使用分布式事务控制
     * (采用的是RocketMQ的事务消息，保证事务的最终一致性)
     * @param orderInfoParam
     */
    @Override
    public void submitOrderInTransaction(OrderInfoParam orderInfoParam) {

        // 1. 扣减秒杀商品库存
        // 1.1 先根据商品id查询秒杀商品库存
        Long skuId = orderInfoParam.getOrderDetailList().get(0).getSkuId();
        SeckillGoodsDTO seckillGoodsDTO = this.getSeckillGoodsDTO(skuId);
        if (seckillGoodsDTO == null){
            /*
                因为有全局异常处理器，所以在这里抛一个业务异常，就可以被全局异常处理器捕获，并且最终返回给前端Result对象
                这个Result对象中，code=217，message=[请求不合法]
             */
            throw new BusinessException(SeckillCodeEnum.SECKILL_ILLEGAL);
        }

        //2. 发送一个事务消息 ——> 之后在事务消息的监听器中，执行本地方法（扣减秒杀库存）
        //上一个方法我们使用seckill_goods表的id和扣减库存的数量进行扣减秒杀库存，这里我们封装一个map作为参数传递到事务生产者
        HashMap<String, Object> localTransationParamMap = new HashMap<>();
        localTransationParamMap.put("id", seckillGoodsDTO.getId());
        localTransationParamMap.put("stock", 1);
        localTransationParamMap.put("skuId", skuId); //这个字段，为了更新本地库存状态位标记
        MqResultEnum mqResultEnum = promoTransactionProducer.sendTransactionMessage(MqTopicConst.PROMO_ORDER_TOPIC, orderInfoParam, localTransationParamMap);
        //在事务消息的消费者那边（订单服务），生成秒杀订单

        //3. 对返回的结果进行判断，因为返回值为void，所有我们通过抛异常来解决
        //3.1 如果返回消息发送失败
        if (mqResultEnum.equals(MqResultEnum.SEND_FAIL)){

            //3.1.1 删除下单的标记
            Long userId = orderInfoParam.getUserId();
            String key = RedisConst.PROMO_USER_ORDERED_FLAG + userId;
            RSet<Long> set = redissonClient.getSet(key);
            set.remove(skuId);

            //3.1.2 返回请稍后再试
            throw new BusinessException(SeckillCodeEnum.SECKILL_ORDER_TRY_AGAIN);
        }

        //3.2 如果返回本地事务执行失败，说明扣减库存失败，也就是没有库存了
        if (MqResultEnum.LOCAL_TRANSACTION_FAIL.equals(mqResultEnum)){

            //返回已售罄
            throw new BusinessException(SeckillCodeEnum.SECKILL_FINISH);
        }

        //3.3 如果返回本地事务执行异常，说明不知道本地事务的执行结果
        if (MqResultEnum.LOCAL_TRANSACTION_EXCEPTION.equals(mqResultEnum)){

            throw new BusinessException(SeckillCodeEnum.SECKILL_ORDER_TRY_AGAIN);
        }

        //3.4 如果返回本地事务执行成功，说明本地库存已经扣减成功，但是消息消费是异步的，此时这个消息有没有被消费到是不确定的，可以返回正在排队中
        throw new BusinessException(SeckillCodeEnum.SECKILL_RUN);

    }
}
